package com.gmail.dengtao.joe.group.transport.impl.filter;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.gmail.dengtao.joe.commons.util.ByteUtils;
import com.gmail.dengtao.joe.transport.filter.FilterAdapter;
import com.gmail.dengtao.joe.transport.filter.FilterEntity;
import com.gmail.dengtao.joe.transport.session.Session;

/**
 * <pre>
 * 1.Filtering out invalid packets.
 * 2.Ensure receiving packet completely.
 * </pre>
 * @author <a href="mailto:joe.dengtao@gmail.com">DengTao</a>
 * @version 1.0
 * @since 1.0
 */
public class DataFilter extends FilterAdapter {

	private Logger LOGGER = LoggerFactory.getLogger(DataFilter.class);

	/** Packet start symbol. */
	public static final byte START_IDENTIFIER = '#';
	/** Packet end symbol. */
	public static final byte END_IDENTIFIER = '!';
	public static final int MAX_PACKET_LEN = 10240000; // 
	
	private Map<Session, byte[]> bufferCacheMap = new HashMap<Session, byte[]>();
	
	@Override
	public void dataReceived(FilterEntity nextEntity, Session session,
			Object data) throws Exception {
		// get buffer...
		byte[] buffer = ByteUtils.EMPTY_BYTE_ARRAY;
		if (bufferCacheMap.containsKey(session)) {
			buffer = bufferCacheMap.get(session);
		}
		// combine buffer...
		buffer = ByteUtils.merge(buffer, (byte[]) data);
		
		// check data...
		if (ByteUtils.isBlank(buffer)) {
			return;
		} else {
			// check data...
			if (ByteUtils.startWith(buffer, START_IDENTIFIER)) {
				do {
					if (buffer.length >= 5) {
						int len = ByteUtils.getInt(buffer, 1);
						if (buffer.length == len) {
							if (ByteUtils.endWith(buffer, END_IDENTIFIER)) {
								super.dataReceived(nextEntity, session, buffer);
							} else {
								// Invalid packet received
								LOGGER.warn("Invalid packet received! Buffer Dump:" + Arrays.toString(buffer) + ", Session:" + session);
							}
							// clear buffer
							bufferCacheMap.put(session, ByteUtils.EMPTY_BYTE_ARRAY);
							break;
						} else if (buffer.length > len) {
							// OK
							byte[] available = ByteUtils.cut(buffer, len);
							if (ByteUtils.endWith(available, END_IDENTIFIER)) {
								super.dataReceived(nextEntity, session, available);
							} else {
								// Invalid packet received
								LOGGER.warn("Invalid packet received! Buffer Dump:" + Arrays.toString(available) + ", Session:" + session);
							}
							// Remain
							buffer = ByteUtils.cut(len, buffer);
						} else {
							// waiting data
							bufferCacheMap.put(session, buffer);
							// Packet not complete
							LOGGER.warn("Packet not complete, waiting data! Buffer Dump:" + Arrays.toString(buffer) + ", Session:" + session);
							break;
						}
					} else {
						// waiting data
						bufferCacheMap.put(session, buffer);
						// Packet not complete
						LOGGER.warn("Packet not complete, waiting data! Buffer Dump:" + Arrays.toString(buffer) + ", Session:" + session);
						break;
					}
				} while (true);
			} else {
				// clear buffer
				bufferCacheMap.put(session, ByteUtils.EMPTY_BYTE_ARRAY);
				// Invalid packet received
				LOGGER.warn("Invalid packet received! Buffer Dump:" + Arrays.toString(buffer) + ", Session:" + session);
			}
		}
	}

	@Override
	public void sessionClosed(FilterEntity nextEntity, Session session)
			throws Exception {
		// remove buffer
		bufferCacheMap.remove(session);
		super.sessionClosed(nextEntity, session);
	}
	
}
